<?php
/**
 * Created by PhpStorm.
 * User: xuaq
 * Date: 2019/1/29
 * Time: 19:23
 */

namespace App\Utils\Queue;


use Enqueue\AmqpExt\AmqpConnectionFactory;
use Enqueue\AmqpExt\AmqpContext;
use Enqueue\AmqpTools\RabbitMqDlxDelayStrategy;
use Interop\Amqp\AmqpTopic;
use Interop\Amqp\Impl\AmqpBind;

class RabbitmqTopicQueue extends BaseTopicQueue
{
    const EXCHANGE = 'php.amqp.ext';

    protected $rabbitTopic = null;
    protected $queue = null;
    protected $context = null;


    /**
     * RabbitmqTopicQueue constructor.
     * 使用依赖注入的方式
     *
     * @param AmqpContext $context
     * @param $exchange
     */
    public function __construct(AmqpContext $context, $exchange)
    {
        $this->rabbitTopic  = $context->createTopic($exchange ?? self::EXCHANGE);
        $this->rabbitTopic->addFlag(AmqpTopic::FLAG_DURABLE);
        $this->rabbitTopic->setType(AmqpTopic::TYPE_FANOUT);
        $context->declareTopic($this->rabbitTopic);
        $this->context = $context;
    }

    public static function getConnection(array $config)
    {
        try {
            $factory = new AmqpConnectionFactory($config);
            $context = $factory->createContext();
            $connection = new self($context, $config['exchange'] ?? null);
        } catch (\Exception $e) {
            echo $e->getMessage() . PHP_EOL;
            return false;
        } catch (\Throwable $e) {
            echo $e->getMessage() . PHP_EOL;
            return false;
        }

        return $connection;
    }

    /**
     * @param string $topic
     * @param JobObject $job
     * @param int $delay 延迟毫秒
     * @param int $priority 优先级
     * @param int $expiration 过期毫秒
     * @throws \Interop\Queue\Exception
     * @throws \Interop\Queue\Exception\Exception
     * @throws \Interop\Queue\Exception\InvalidDestinationException
     * @throws \Interop\Queue\Exception\InvalidMessageException
     * @throws \Interop\Queue\Exception\PriorityNotSupportedException
     * @throws \Interop\Queue\Exception\TimeToLiveNotSupportedException
     */
    public function push(string $topic, JobObject $job, int $delay = 0, int $priority = BaseTopicQueue::HIGH_LEVEL_1, int $expiration=0)
    {
        if (!$this->isConnected()) {
            return;
        }

        $queue = $this->createQueue($topic);
        $this->context->bind(new AmqpBind($this->rabbitTopic, $queue));
        $message = $this->context->createMessage(serialize($job));
        $producer = $this->context->createProducer();

        if ($delay > 0) {
            $producer->setDelayStrategy(new RabbitMqDlxDelayStrategy())
                ->setDelayStrategy($delay);
        }

        if ($priority) {
            $producer->setPriority($priority);
        }

        if ($expiration > 0) {
            $producer->setTimeToLive($expiration);
        }

        return $producer->send($queue, $message);
    }

    public function pop($topic)
    {
        if (!$this->isConnected()) {
            return;
        }

        $queue = $this->createQueue($topic);
        $consumer = $this->context->createConsumer($queue);
        if ($m = $consumer->receive(1)) {
            $result = $m->getBody();
            $consumer->acknowledge($m);
        }

        return !empty($result) ? unserialize($result) : null;
    }

    public function len($topic)
    {
        if (!$this->isConnected()) {
            return;
        }

        $queue = $this->createQueue($topic);
        $len   =$this->context->declareQueue($queue);

        return $len ?? 0;
    }

    public function close()
    {
        if (!$this->isConnected()) {
            return;
        }

        $this->context->close();
    }

    public function isConnected()
    {
        return $this->context->getExtChannel()->getConnection()->isConnected();
    }

    private function createQueue($topic)
    {
        try {
            $queue = $this->context->createQueue($topic);
            $queue->addFlag(AmqpTopic::FLAG_DURABLE);
            $queue->setArgument('x-max-length', 10);
            //执行创建队列
            $this->context->declareQueue($queue);
        } catch (\Exception $e) {
            echo $e->getMessage() . PHP_EOL;
            return false;
        } catch (\Throwable $e) {
            echo $e->getMessage() . PHP_EOL;
            return false;
        }
        return $queue;
    }

}